package pl.edu.agh.broker.consumer;

import java.io.FileInputStream;
import java.util.Properties;

import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.ObjectMessage;
import javax.jms.Session;
import javax.jms.Topic;
import javax.jms.TopicConnection;
import javax.jms.TopicSession;
import javax.jms.TopicSubscriber;

import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.log4j.Logger;
import org.apache.log4j.PropertyConfigurator;

import pl.edu.agh.broker.mns.client.MultiNameServiceLocator;
import pl.edu.agh.broker.mns.shared.MultiNameService;
import pl.edu.agh.stock.data.IndexDataPoint;

public class Main {

	private static final String INDEXES = "indexes";
	private static final String MNS_ADDRESS = "mns.address";
	private static final String MNS_PORT = "mns.port";
	private static final String BROKER_TOPIC = "broker.topic";

	private static final String LOGGER_CONFIG = "src/main/resources/log4j.properties";
	private static final Logger LOGGER = Logger.getLogger(Main.class);
	private static final String PROPERTY_FILE = "conf/consumer.properties";

	public static void main(String[] args) {

		// initialize log4j
		PropertyConfigurator.configure(LOGGER_CONFIG);

		try {
			// load properties
			Properties properties = new Properties();
			properties.load(new FileInputStream(PROPERTY_FILE));

			// resolve proxy brokers addresses
			MultiNameService mns = MultiNameServiceLocator.getMultiNameService(
					properties.getProperty(MNS_ADDRESS),
					Integer.parseInt(properties.getProperty(MNS_PORT)));

			for (String index : properties.getProperty(INDEXES).split(",")) {

				String brokerURL = (String) mns.getAny(index);
				if (brokerURL == null) {
					continue;
				}
				ActiveMQConnectionFactory localConnectionFactory = new ActiveMQConnectionFactory(
						brokerURL);
				TopicConnection connection = localConnectionFactory
						.createTopicConnection();
				connection.start();
				TopicSession session = connection.createTopicSession(false,
						Session.AUTO_ACKNOWLEDGE);
				Topic topic = session.createTopic(properties
						.getProperty(BROKER_TOPIC));
				TopicSubscriber topicSubscriber = session
						.createSubscriber(topic);

				topicSubscriber.setMessageListener(new MessageListener() {
					@Override
					public void onMessage(Message message) {
						try {
							if (message instanceof ObjectMessage) {
								ObjectMessage objectMessage = (ObjectMessage) message;
								IndexDataPoint stockDataPoint = (IndexDataPoint) objectMessage
										.getObject();
								LOGGER.info("Received index: "
										+ stockDataPoint.getName() + " value "
										+ stockDataPoint.getValors());
							} else {
								LOGGER.info("Invalid message received.");
							}
						} catch (JMSException e) {
							LOGGER.warn("", e);
						}
					}
				});
			}
			
			LOGGER.info("Stock Consumer started");

			while (true) {
				// do nothing
			}

		} catch (Exception e) {
			LOGGER.error("", e);
		}
	}

}
